package com.hou.rocketmq.producer;

import com.hou.rocketmq.pojo.User;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.messaging.support.MessageBuilder;

import java.util.List;

public class DefaultMQProducerTest {

    public static void main(String[] args) {
        DefaultMQProducer producer = new DefaultMQProducer("hou");//group
        producer.setInstanceName("instance1");// 如果一个jvm有多个produce,就需要设置区分
        producer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        producer.setRetryTimesWhenSendAsyncFailed(0);//设置重试次数,保证不丢失
        for (int i = 0; i < 10; i++) {
            try {
                final int index = i;
                Message msg = new Message("test-topic-01",   //topic
                        "TagA", //发送消息时候用来设置可过滤的标签
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 仅支持预设值的时间长度 (1s/5s/1Os/30s/Im/2m/3m/4m/5m/6m/ 7m/8m/9m/1Om/20m/30m/1h/2h)
                //msg.setDelayTimeLevel(3); // 3  对应 10s  ,   发送延迟消息
                //发送时候传入MessageQueueSelector实现可自定义发送规则
                producer.send(msg, (List<MessageQueue> mqs, Message message, Object arg)->{
                    return mqs.get(0);
                },new SendCallback() {
                    @Override  // 成功
                    public void onSuccess(SendResult sendResult) {
                        //发送结果状态,一共有四种
                        final SendStatus sendStatus = sendResult.getSendStatus();
                        switch (sendStatus){
                            case SEND_OK:  // 发送成功
                                break;
                            //表示没有再规定的时间内完成刷盘,同步刷盘策略才会出现此状态
                            case FLUSH_DISK_TIMEOUT:
                                break;
                            // 表示在主从模式下,,并且broker是同步master,没有在设定时间内完成主从同步
                            case FLUSH_SLAVE_TIMEOUT:
                                break;
                            //主从模式下,broker也是同步maser,没有找到salve的broker
                            case SLAVE_NOT_AVAILABLE:
                                break;
                        }
                        System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                    }
                    @Override   // 异常回调
                    public void onException(Throwable e) { e.printStackTrace(); }
                });
            } catch (Exception e) { e.printStackTrace(); }
        }
        //producer.shutdown(); 不能先关闭,否则会直接断开
    }
}
